import com.zuikaku.pojo.Order;
import com.zuikaku.sink.OrderSink;
import com.zuikaku.source.OrderSource;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class SinkDemo {
    public static void main(String[] args) {
        //Flink流处理案例
        //取得流运行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
        env.setParallelism(2);
        //自定义流Source
        DataStream<Order> orderSource = env.addSource(new OrderSource());
        orderSource.print("源数据(source)order");
        orderSource.filter(new FilterFunction<Order>() {
            @Override
            public boolean filter(Order value) throws Exception {
                //暂时不做筛选，所有的都通过
                return true;
            }
        });
        //添加自定义Sink进行消费
        orderSource.addSink(new OrderSink());

        try {
            env.execute("sink api demo");
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}
